package com.douxg.learning.bigdata.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

@Component
public class KafkaConsumer {

    @KafkaListener(groupId = "one", topics = {"flow"}, containerFactory = "kafkaListenerContainerFactory")
    public void consumer(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
//        System.out.println("size:" + consumerRecords.size());
//        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            //判断是否为null
            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            if (kafkaMessage.isPresent()) {
                //得到Optional实例中的值
                Object message = kafkaMessage.get();
                System.err.println(System.currentTimeMillis() + "消费消息:" + message);
            }
//        }
        acknowledgment.acknowledge();
    }
}
